Amplify Gen2 で DynamoDB Streams を使って DynamoDB テーブル間のデータ連携を行う

import { type ClientSchema, a, defineData } from '@aws-amplify/backend';

const schema = a.schema({
  ScoreActivities: a
    .model({
      team_id: a.string().required(),
      score_change: a.integer().required(),
    })
    .authorization((allow) => [
      allow.ownerDefinedIn("team_id").identityClaim("custom:team_id"),
      allow.publicApiKey()
    ])
});

:

import { type ClientSchema, a, defineData } from '@aws-amplify/backend';

const schema = a.schema({
  ScoreActivities: a
    .model({
      team_id: a.string().required(),
      score_change: a.integer().required(),
    })
    .authorization((allow) => [
      allow.ownerDefinedIn("team_id").identityClaim("custom:team_id"),
      allow.publicApiKey()
    ]),
  TotalScore: a
    .model({
      team_id: a.string().required(),
      total_score: a.integer().required(),
    })
    .identifier(['team_id'])
    .authorization((allow) => [
      allow.ownerDefinedIn("team_id").identityClaim("custom:team_id"),
      allow.publicApiKey()
    ]),
});

:

import { defineBackend } from '@aws-amplify/backend';
import { auth } from './auth/resource';
import { data } from './data/resource';
import { preTokenGenerationV2 } from './auth/pre-token-generation-v2/resource';
import { Effect, PolicyStatement } from 'aws-cdk-lib/aws-iam';
import { EventSourceMapping, StartingPosition } from 'aws-cdk-lib/aws-lambda';

import { NodejsFunction} from 'aws-cdk-lib/aws-lambda-nodejs';
import * as url from 'node:url';

const backend = defineBackend({
  auth,
  data,
  preTokenGenerationV2,
});

const { cfnUserPool } = backend.auth.resources.cfnResources
cfnUserPool.addPropertyOverride("LambdaConfig.PreTokenGenerationConfig",{
  LambdaVersion: 'V2_0',
  LambdaArn: backend.preTokenGenerationV2.resources.lambda.functionArn,
});

const updateTotalScoreStack = backend.createStack('UpdateTotalScoreStack');
const funcitonUpdateTotalScoreOnActivity = new NodejsFunction(
  updateTotalScoreStack,
  'update-totalscore-on-activity',
  {
    entry: url.fileURLToPath(new URL('./functions/update-totalscore-on-activity/handler.ts', import.meta.url)),
    environment: {
      APPSYNC_ENDPOINT: backend.data.resources.cfnResources.cfnGraphqlApi.attrGraphQlUrl,
      APPSYNC_API_KEY: backend.data.resources.cfnResources.cfnApiKey?.attrApiKey || '',
    },
    initialPolicy: [
      new PolicyStatement({
        effect: Effect.ALLOW,
        actions: [
          "dynamodb:DescribeStream",
          "dynamodb:GetRecords",
          "dynamodb:GetShardIterator",
          "dynamodb:ListStreams",
        ],
        resources: ["*"],
      }),
    ],
  }
)
new EventSourceMapping(
  updateTotalScoreStack,
  "TotalScoreFunctionEventSourceMapping",
  {
    eventSourceArn: backend.data.resources.tables["ScoreActivities"].tableStreamArn,
    target: funcitonUpdateTotalScoreOnActivity,
    batchSize: 100,
    startingPosition: StartingPosition.LATEST,
  }
);

import axios, { AxiosError } from "axios";
import type { DynamoDBStreamHandler } from "aws-lambda";
import { Logger } from "@aws-lambda-powertools/logger";

type TotalScoreInput = {
  team_id: string;
  total_score: number;
};

type GraphQLError = {
  message: string;
  path: string[];
  errorType: string;
};

const logger = new Logger({
  logLevel: "INFO",
  serviceName: "dynamodb-stream-handler",
});

// GraphQL操作の定義
const CREATE_TOTAL_SCORE = `
  mutation CreateTotalScore($input: CreateTotalScoreInput!) {
    createTotalScore(input: $input) {
      team_id
      total_score
    }
  }
`;

const UPDATE_TOTAL_SCORE = `
  mutation UpdateTotalScore($input: UpdateTotalScoreInput!) {
    updateTotalScore(input: $input) {
      team_id
      total_score
    }
  }
`;

const GET_TOTAL_SCORE = `
  query GetTotalScore($team_id: String!) {
    getTotalScore(team_id: $team_id) {
      team_id
      total_score
    }
  }
`;

async function executeGraphQLRequest(query: string, variables: any) {
  if (!process.env.APPSYNC_ENDPOINT || !process.env.APPSYNC_API_KEY) {
    throw new Error("Missing required environment variables: APPSYNC_ENDPOINT or APPSYNC_API_KEY");
  }

  const response = await axios.post(
    process.env.APPSYNC_ENDPOINT,
    {
      query,
      variables,
    },
    {
      headers: {
        'Content-Type': 'application/json',
        'x-api-key': process.env.APPSYNC_API_KEY,
      },
    }
  );

  if (response.data.errors) {
    throw new Error(`GraphQL Error: ${JSON.stringify(response.data.errors)}`);
  }

  return response.data;
}

async function updateOrCreateTotalScore(teamId: string, newTotalScore: number) {
  try {
    const updateResponse = await executeGraphQLRequest(
      UPDATE_TOTAL_SCORE,
      {
        input: {
          team_id: teamId,
          total_score: newTotalScore
        }
      }
    );
    logger.info('Successfully updated total score', { teamId, newTotalScore });
    return updateResponse;
  } catch (error) {
    if (error instanceof Error) {
      if (error.message.includes('ConditionalCheckFailedException')) {
        logger.info('Record not found, creating new total score', { teamId, newTotalScore });
        const createResponse = await executeGraphQLRequest(
          CREATE_TOTAL_SCORE,
          {
            input: {
              team_id: teamId,
              total_score: newTotalScore
            }
          }
        );
        return createResponse;
      }
    }
    throw error;
  }
}

export const handler: DynamoDBStreamHandler = async (event) => {
  const batchItemFailures: Array<{itemIdentifier: string}> = [];
  
  logger.info('Starting to process DynamoDB Stream records', { 
    recordCount: event.Records.length 
  });

  for (const record of event.Records) {
    try {
      if (record.eventName !== 'INSERT') {
        logger.info(`Skipping non-INSERT event: ${record.eventName}`);
        continue;
      }

      const newImage = record.dynamodb?.NewImage;
      if (!newImage) {
        logger.warn('No NewImage in record, skipping');
        continue;
      }

      const teamId = newImage.team_id?.S;
      const scoreChange = parseInt(newImage.score_change?.N ?? '0');

      if (!teamId || scoreChange === 0) {
        logger.warn('Invalid record data', { teamId, scoreChange });
        continue;
      }

      const getTotalScoreResponse = await executeGraphQLRequest(
        GET_TOTAL_SCORE,
        { team_id: teamId }
      );

      const currentScore = getTotalScoreResponse.data.getTotalScore?.total_score ?? 0;
      const newTotalScore = currentScore + scoreChange;

      const result = await updateOrCreateTotalScore(teamId, newTotalScore);

      logger.info('Successfully processed score change', {
        teamId,
        scoreChange,
        currentScore,
        newTotalScore,
        operation: result.data.createTotalScore ? 'create' : 'update'
      });

    } catch (error) {
      const errorMessage = error instanceof Error ? error.message : String(error);
      const errorStack = error instanceof Error ? error.stack : undefined;

      logger.error('Error processing record', {
        error: errorMessage,
        stack: errorStack,
        recordId: record.eventID
      });
      
      if (record.eventID) {
        batchItemFailures.push({
          itemIdentifier: record.eventID
        });
      }
    }
  }
  
  return { batchItemFailures };
};